package com.ruixinzheng.rxjavademo;

import android.annotation.SuppressLint;
import android.os.SystemClock;
import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.Notification;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.core.SingleEmitter;
import io.reactivex.rxjava3.core.SingleObserver;
import io.reactivex.rxjava3.core.SingleOnSubscribe;
import io.reactivex.rxjava3.core.SingleSource;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.subjects.PublishSubject;
import io.reactivex.rxjava3.subjects.Subject;

@SuppressLint("CheckResult")
public class RxJavaDemo {
    private final String TAG = RxJavaDemo.class.getSimpleName();
    
    public void switchThread() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                Thread th = Thread.currentThread();
                Log.i(TAG, "emitter: " + th.getName() + " " + th.getId());
                emitter.onNext("hello");
                emitter.onComplete();
            }
        })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .doOnEach(new Consumer<Notification<String>>() {
                    @Override
                    public void accept(Notification<String> stringNotification) throws Throwable {
                        String value = stringNotification.getValue();
                        Thread th = Thread.currentThread();
                        Log.i(TAG, "doOnEach: " + value + " " + th.getName() + " " + th.getId());
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Thread th = Thread.currentThread();
                        Log.i(TAG, "subscribe: " + s + " " + th.getName() + " " + th.getId());
                    }
                });
    }

    // 优先使用内存缓存，没有就请求网络
    public void concat() {
        final Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                String memoryCache = new Random().nextBoolean() ? "memory" : null;
                if (memoryCache != null) {
                    emitter.onNext(memoryCache);
                } else {
                    emitter.onComplete();
                }
            }
        });

        Observable<String> network = Observable.just("network");

        Observable.concat(memory, network).firstElement()
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, "accept: " + s);
                    }
                });
    }

    // 并发执行多个请求
    public void merge() {
        Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("first request");
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("second request");
            }
        }).subscribeOn(Schedulers.io());

        Observable.merge(observable1, observable2)
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });
    }

    // 并发执行多个请求，且可对各个请求结果进行合并处理
    public void zip(){
        Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("first request");
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("second request");
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<String, String, String>() {
                    @Override
                    public String apply(String s, String s2) throws Throwable {
                        return s + " " + s2;
                    }
                })
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });
    }

    // 使用map转换数据
    public void map() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer + " time";
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, "accept: " + s);
            }
        });
    }

    // 可用于嵌套网络请求，利用前一个请求的结果发起另一个请求
    public void flatMap() {
        Single.create(new SingleOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull SingleEmitter<String> emitter) throws Throwable {
                emitter.onSuccess("result1");
            }
        }).flatMap(new Function<String, SingleSource<String>>() {
            @Override
            public SingleSource<String> apply(String s) throws Throwable {
                return Single.create(new SingleOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull SingleEmitter<String> emitter) throws Throwable {
                        emitter.onSuccess(s + " result2");
                    }
                });
            }
        }).subscribeOn(Schedulers.io()).subscribe(new SingleObserver<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onSuccess(@NonNull String s) {
                Log.i(TAG, "onSuccess: " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }
        });

    }

    // concatMap和flatMap的区别是前者转换成的多个Observable的执行是有序的
    public void concatMap() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("a");
                emitter.onNext("b");
                emitter.onNext("c");
            }
        }).concatMap(new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(String s) throws Throwable {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add(s + i);
                }
                return Observable.fromIterable(list).subscribeOn(Schedulers.newThread());
            }
        })
                .subscribeOn(Schedulers.io())
                //.observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });
    }

    // 500ms超时后数据才会被发射，发射数据后超时时间重新计算
    public void debounce() {
        final Subject<String> subject = PublishSubject.create();

        subject.debounce(500, TimeUnit.MILLISECONDS)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });

        new Thread() {
            @Override
            public void run() {
                super.run();
                subject.onNext("a");
                SystemClock.sleep(700);
                subject.onNext("ab"); // 这个数据被忽略，因为100ms后有新数据
                SystemClock.sleep(100);
                subject.onNext("abc");
            }
        }.start();
    }

    // observable可发射多个数据
    public void observable() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull String s) {
                Log.i(TAG, "onNext: " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete: ");
            }
        });
    }

    // Observable可以发射多个数据，而Single只发射一个数据
    public void single() {
        Single.create(new SingleOnSubscribe<String>() {
            @Override
            public void subscribe(SingleEmitter<String> e) throws Exception {
                e.onSuccess("hello world");
            }
        }).subscribe(new SingleObserver<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onSuccess(String value) {
                Log.i(TAG, "onSuccess: " + value);
            }

            @Override
            public void onError(Throwable e) {

            }
        });
    }

    // 使用subject可以实现在不同的地方发射数据和接收数据
    public void subject() {
        final Subject<String> subject = PublishSubject.create();

        subject.subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });

        new Thread() {
            @Override
            public void run() {
                super.run();
                subject.onNext("a");
                subject.onNext("b");
                subject.onNext("c");
            }
        }.start();
    }

    // Flowable和Observable用法差不多，但Flowable可设定背压策略，以及允许下游控制上游发射的数量，适合有大量数据的场景
    public void flowable() {
        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<String> emitter) throws Throwable {
                emitter.onNext("a");
                emitter.onNext("b");
                emitter.onNext("c");
            }
        }, BackpressureStrategy.ERROR).subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, "onNext: " + s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    // 使用just可以很简洁地发射多个数据
    public void just() {
        Observable observable = Observable.just("you", "are", "wonderful");
        observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, "accept: " + s);
            }
        });
    }

    // 使用from发射数组
    public void from() {
        Observable.fromArray(new Integer[]{1, 2, 3, 4, 5}).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "accept: " +integer);
            }
        });
    }

    // 过滤数据
    public void filter() {
        Observable.fromArray(new Integer[]{1, 2, 3, 4, 5})
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer % 2 == 1;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: " + integer);
                    }
                });
    }

    // take指定获取多少个数据
    public void take() {
        Observable.just("1", "2", "3", "4", "5", "6")
                .take(2)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.i(TAG, "accept: " + s);
                    }
                });
    }

    // doOnNext可在观察者获得数据前执行其他工作
    public void doOnNext() {
        Observable.fromArray(new Integer[]{1, 2, 3})
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "doOnNext: " + integer);
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        Log.i(TAG, "accept: " + integer);
                    }
                });
    }

    // 使用Disposable.dispose()停止接收数据
    public void disposable() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "emitter 1");
                emitter.onNext(1);
                Log.i(TAG, "emitter 2");
                emitter.onNext(2);
                Log.i(TAG, "emitter 3");
                emitter.onNext(3);
                Log.i(TAG, "emitter complete");
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.i(TAG, "onNext: " + value);
                i++;
                if (i == 2) {
                    Log.i(TAG, "dispose");
                    mDisposable.dispose();
                    Log.i(TAG, "isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "error");
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "complete");
            }
        });
    }
}
